package pipeline

import (
	"encoding/binary"
	"fmt"
	"io"
	"math/rand"
	"sort"
	"time"
)

var startTime time.Time
func Init()  {
	startTime=time.Now()
}

//将数据依次压入管道(读取数据)
func ArraySource(a ...int) <-chan int {
	out := make(chan int)

	go func() {
		for _, v := range a {
			out <- v
		}

		close(out)
	}()

	return out
}

//内存中排序(内部排序)
func InMemSort(in <-chan int) <-chan int {
	out := make(chan int,1024)
	go func() {
		//接收管道数据
		a := []int{}
		for v := range in {
			a = append(a, v)
		}

		fmt.Println("Read done:",time.Now().Sub(startTime))

		//排序所有来自in管道中的数据
		sort.Ints(a)

		fmt.Println("InMemSort done:",time.Now().Sub(startTime))
		//将排序后的数据压入新的管道
		for _, v := range a {
			out <- v
		}

		close(out)
	}()

	return out
}


//将两根排好序的管道进行归并
func Merge(in1, in2 <-chan int) <-chan int {
	out := make(chan int,1024)
	go func() {
		v1, ok1 := <-in1
		v2, ok2 := <-in2
		for ok1 || ok2 {

			if (ok1 &&v1<= v2) || !ok2 {
				out<-v1
				v1,ok1=<-in1
			}else{
				out<-v2
				v2,ok2=<-in2
			}
		}
		fmt.Println("Merge done:",time.Now().Sub(startTime))

		close(out)
	}()

	return out
}

//多路两两归并
func MergeN(inputs ...<-chan int)<-chan int {
	if len(inputs)== 1 {
		return inputs[0]
	}

	m:=len(inputs)/2

	//merge inputs[0...m) and inputs [m ..end)
	return Merge(MergeN(inputs[0:m]...),MergeN(inputs[m:]...))
}


//读取数据
func ReaderSource(reader io.Reader,chunkSize int) <-chan int  {
	out:=make(chan int,1024)
	go func() {
		buffer:=make([]byte,8)
		bytesRead:=0
		for {
			n,err:=reader.Read(buffer)
			bytesRead+=n
			if n> 0 {
				v:=int(binary.BigEndian.Uint64(buffer))

				out<-v
			}

			if err != nil ||(chunkSize!=-1 && bytesRead>=chunkSize) {
				break
			}
		}

		close(out)
	}()

	return out
}


//接收数据,写入
func WriterSink(writer io.Writer,in <-chan int) {
	//收channel数据
	for v:=range in{
		buffer:=make([]byte,8)
		binary.BigEndian.PutUint64(buffer,uint64(v))

		writer.Write(buffer)
	}
}

//生成随机数据,num 决定生成多少个8字节的数
func RandomSource(num int) <-chan int {
	out:=make(chan int)
	go func() {
		for i:=0;i<num;i++{
			out<-rand.Int()
		}

		close(out)
	}()
	return out
}

